package com.atguigu.bigdata.spark.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function0;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

public class SparkStreaming09_Resume_JAVA {
    public static void main(String[] args) throws InterruptedException {
        JavaStreamingContext jssc = JavaStreamingContext.getOrCreate("cp", new Function0<JavaStreamingContext>() {
            @Override
            public JavaStreamingContext call() throws Exception {
                SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming");
                JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(3));
                JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost",9999);
                JavaPairDStream<String, Integer> wordToMap = lines.mapToPair(new PairFunction<String, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(String s) throws Exception {
                        return new Tuple2<String, Integer>(s, 1);
                    }
                });

                wordToMap.print();
                return jssc;
            }
        });

        jssc.checkpoint("cp");

        jssc.start();

        jssc.awaitTermination();
    }
}
